package org.jetlinks.pro.device.service.task;

import lombok.Generated;
import lombok.Getter;
import lombok.NonNull;
import lombok.Setter;
import org.jetlinks.core.cluster.ClusterManager;
import org.jetlinks.core.device.DeviceOperator;
import org.jetlinks.core.message.*;
import org.jetlinks.core.message.interceptor.DeviceMessageSenderInterceptor;
import org.jetlinks.pro.PropertyConstants;
import org.jetlinks.pro.utils.MessageTypeMatcher;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@Component
@Generated
public class DeviceMessageSendTaskInterceptor implements DeviceMessageSenderInterceptor {

    private final DeviceMessageTaskRepository taskRepository;

    private final String serverId;

    @Getter
    @Setter
    private boolean defaultEnabled = false;

    @Getter
    @Setter
    @NonNull
    private MessageTypeMatcher type = new MessageTypeMatcher();

    public DeviceMessageSendTaskInterceptor(DeviceMessageTaskRepository repository, ClusterManager clusterManager) {
        type.includes(MessageType.INVOKE_FUNCTION);
        this.taskRepository = repository;
        this.serverId = clusterManager.getCurrentServerId();
    }

    @Override
    public <R extends DeviceMessage> Flux<R> afterSent(DeviceOperator device,
                                                       DeviceMessage message,
                                                       Flux<R> reply) {
        message.addHeader(PropertyConstants.serverId, serverId);
        if (message.getHeader(PropertyConstants.useTask).orElse(false)
            || (defaultEnabled && type.match(message.getMessageType()))) {

            //转发的子设备消息不处理
            if(message instanceof ChildDeviceMessage&&
                ((ChildDeviceMessage) message).getChildDeviceMessage().getHeader(Headers.dispatchToParent).orElse(false)){
                return reply;
            }

            String taskId = message.getHeaderOrDefault(PropertyConstants.taskId);

            //回复结果处理逻辑
            Flux<R> replyHandle = reply
                .flatMap(response -> {
                    response.addHeader(PropertyConstants.taskId, message.getHeaderOrDefault(PropertyConstants.taskId));
                    if (response instanceof DeviceMessageReply) {
                        return this
                            .taskRepository
                            .updateResult(DeviceMessageResponse.of(message,((DeviceMessageReply) response)))
                            .thenReturn(response);
                    } else {
                        return Mono.just(response);
                    }
                })
                .onErrorResume(err -> this
                    .taskRepository
                    .updateResult(DeviceMessageResponse.of(message, err))
                    .then(Mono.error(err))
                );
            //如果包含taskId,则直接处理回复,无需重复创建任务.
            if (StringUtils.hasText(taskId)) {
                return replyHandle;
            }
            //创建任务后再处理回复.
            return taskRepository
                .addTask(DeviceMessageTask.newTask(message))
                .thenMany(replyHandle);
        }

        return reply;
    }

    @Override
    public int getOrder() {
        return Integer.MAX_VALUE;
    }
}
